package com.cetcs.kmga.dataManager.dao.hbase.impl;

import com.cetcs.kmga.dataManager.util.HBasePageModel;
import com.cetcs.kmga.dataManager.util.HBasePool;
import com.google.common.collect.Lists;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.List;
import java.util.concurrent.Callable;
import java.util.stream.Collectors;

/**
 * 多线程查询探针日志
 *
 * @author xutao
 * @version V1.0 创建时间：2017-12-13 16:26
 *          Copyright 2017 by CETC
 */
public class ProbFuzzyFilterQuery implements Callable<List<String>> {

    private String starKey;

    private String endKey;

    private String tbName;

    private FilterList filterList;

    private int pageSize;

    public ProbFuzzyFilterQuery(String starKey, String endKey, String tbName, FilterList filterList, int pageSize) {
        this.starKey = starKey;
        this.endKey = endKey;
        this.tbName = tbName;
        this.filterList = filterList;
        this.pageSize = pageSize;
    }

    @Override
    public List<String> call() throws Exception {
        HBasePageModel ret = new HBasePageModel(pageSize);
        ret = HBasePool.getInstance( ).
                scanResultByPageFilter(tbName, null, starKey.getBytes( ), endKey.
                        getBytes( ), filterList, 0, ret);
        List<Result> indexResultList = ret.getResultList( );
        List<String> keyRet = Lists.newArrayList( );
        for (int i = 0; i < indexResultList.size( ); i++) {
            Result result = indexResultList.get(i);
            for (Cell cell : result.listCells( )) {
                String key = Bytes.toString(CellUtil.cloneQualifier(cell));
                if (key.equalsIgnoreCase("ROWKEY")) {
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    keyRet.add(value);
                }
            }
        }
        List<String> newKeyRet = keyRet.stream( ).distinct( ).collect(Collectors.toList( ));
        return newKeyRet;
    }
}
